package com.yupi.springbootinit.mq;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;

/**
 * 多消费者案例
 */
public class MultiConsumer {

    private static final String TASK_QUEUE_NAME = "multi_queue";

    public static void main(String[] argv) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        final Connection connection = factory.newConnection();

        //创建两个消费者
        for (int finalI = 0; finalI < 2; finalI++) {
            final Channel channel = connection.createChannel();

            //声明队列，队列持久化， 服务器重启队列不消失
            channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);
            System.out.println(" [*] Waiting for messages. To exit press CTRL+C");

            //控制单个消费者的处理任务的积压数
            //每个消费者最多同时处理1个任务
            channel.basicQos(1);

            //这里定义如何处理消息
            int finalI1 = finalI;
            DeliverCallback deliverCallback = (consumerTag, delivery) -> {
                String message = new String(delivery.getBody(), "UTF-8");

                try {
                    //处理工作
                    System.out.println(" [x] Received '" + "编号：" + finalI1 + message + "'");
                    //消息确认                                          这里设置为false是为了实现一条消息一次确认，防止批量确认
                    channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);

                    //停20秒，模拟机器处理能力有限
                    Thread.sleep(20000);

                } catch (InterruptedException e) {
                    e.printStackTrace();
                    //消息拒绝                                                                     是否需要再放回队列
                    channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, false);
                } finally {
                    System.out.println(" [x] Done");
                    channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
                }
            };
            //开启消费监听                          false关闭自动消息确认。建议自己手动确认
            channel.basicConsume(TASK_QUEUE_NAME, false, deliverCallback, consumerTag -> { });
        }
    }
}
